Field
A computational grid (also known as a compute cluster) consists of a set of computers interconnected by a computer network which are used to run a single application program which is often called a parallel application. Each computer is usually called a computational node. The application program is subdivided into computational tasks, each of which is run on a different node, and each of which is a portion of a computer program. These tasks usually operate simultaneously on distinct data, and the tasks may communicate with each other as part of the overall computation performed by the application.
One example of a parallel application is a weather simulation model in which each node of the computational grid runs a task that models weather effects in a distinct region of the world at a given simulated time of day. All of the tasks together model the entire weather system for the combined regions at the simulated time. As part of each task, the node may communicate with other nodes to obtain or deliver weather information from a neighboring region. The parallel application models the weather over a sequence of time steps.
A common method for initiating the tasks that comprise the parallel application is to perform a data parallel operation in which one node instructs all nodes to perform a single computational function (sometimes called a method) on distinct sets of data (sometimes called objects). A data-parallel operation consists of a set of method invocations on selected data within most or all of the nodes of the computational grid. To maximize performance, these method invocations are run in parallel (that is, simultaneously) on the computational nodes All nodes execute this function and either produce new data or update their specific sets of data. As part of the implementation of the data parallel operation, the initiating node is informed when all nodes have finished running this method so that it can start another data parallel operation. An example of a data parallel operation is shown in FIG. 2.
For example, the objects which represent the above weather simulation model could be denoted as a list of region objects, region[r], for a set of regions r, and the computational method could be denoted by the function model_weather( ); this function operates on a region object. We can assume that the region objects have been partitioned among the nodes of the computational grid by some means (such as the use of a distributed cache). In a data parallel operation, one node instructs all nodes to run the model_weather( )method on their assigned regions.
Another example of a parallel application is a financial analysis application in which each node of the computational grid runs a task that analyzes a distinct set of financial portfolios and collects results for each portfolio. For example, the analysis might analyze each portfolio based on current market conditions to determine the portfolio's value. The results of each analysis are then combined into a final report covering all portfolios.
In the financial analysis example, the objects which represent the above financial portfolios could be denoted as a list of portfolio objects, portfolio[c], for a set of customers c, and the computational method could be denoted by the function analyze_portfolio( ); this function operates on a portfolio object. We can assume that the portfolio objects have been partitioned among the nodes of the computational grid by some means (such as the use of a distributed cache). In a data parallel operation, one node instructs all nodes to run the analyze_portfolio( )method on their locally stored portfolio objects. By doing so, the computational grid analyzes each node's subset of the portfolio objects in parallel to minimize the completion time of the operation.
Data parallel operations are distinguished from another method for initiating tasks, called control parallel operations (also called the manager/worker paradigm) in which the nodes within the grid repeatedly seek and receive independent work tasks from a manager node. Although both methods have uses in various applications, data parallel applications are the subject of the present invention.
The implementation of a data parallel operation requires that the following three steps be performed in sequence:                1. One node initiates a task on all nodes in order to invoke a specified method on multiple data sets.        2. All nodes perform the task by executing the specified method on distinct data sets.        3. All nodes communicate to the originating node (or its successor if a failure occurs) that the task has been completed on all specified data sets and report results which are merged together.        
Data parallel operations have been implemented for numerous parallel computing systems for several decades. However, prior implementations usually have assumed that computational nodes and the communications between nodes do not fail during the execution of the data parallel operation. If a failure occurs, the data parallel operation is restarted. Unfortunately, many time-critical applications, such as financial services, cannot tolerate the delay required to re-run a data parallel operation.
An implementation of a data parallel operation that survives the failure of a server or its portion of the communications network is said to be a highly available implementation. The present invention describes a method to perform a highly available data parallel operation so that it can be successfully completed even if a node or a portion of the network (such as network interface card) fails during one of the three steps described above.
Description of Related Art
Numerous computational grids (also known as parallel computing systems) have been created over the last few decades to implement data parallel operations. Two examples include the Connection Machine from Thinking Machines, the Intel Paragon Parallel Supercomputer. More recently, computational grids have been implemented as clusters of server computers with data parallel operations implemented in software using standard TCP/IP communications networks. However, none of these implementations handles the problem of making data parallel operations highly available after a server or partial network failure.
Reliable, distributed computing systems have tackled the problem of creating a membership for a set of cooperating computers and making that membership highly available after a server fails. These systems provide a software layer that runs on all the cooperating computers and allows the computers to join and leave the common membership. They implement a form of reliable multicast, so all computers can be sure to have received a multicast message. If a computer should fail, the software forms a new membership and identifies the new membership to all surviving members. Member computers can send messages to one or more computers and can usually determine which nodes in each membership that have received their messages. Two examples of these distributed computing systems are Isis and Ensemble.
By using reliable multicast and detecting membership changes if a failure occurs, reliable, distributed computing systems have produced a method for implementing step 1 of a data parallel operation in a highly available manner. The black arrows in FIG. 2 illustrate the use of a reliable multicast to distribute a method invocation to all nodes. However, these systems have not implemented all three steps in a data parallel operation. In particular, they do not provide a means to ensure that the method is executed on all data sets and that the originating node (or its successor if a failure occurs) is notified of completion.
Because reliable, distributed computing systems are focused primarily on computer memberships, they do not provide a means to track the data assigned to each computer for the purposes of completing a data parallel operation. Knowing the new membership after a computer failure does not tell the surviving nodes which data sets the method has completed and which data sets still need the method executed on them. It also does not provide a means for managing the placement of data sets on the nodes so that the surviving nodes can re-distribute the tasks that would have been performed by the failed nodes.
Distributed caches, such as commercial software products from ScaleOut Software, Inc. and Tangosol, have been created to store data objects and make them available to all nodes in a computational grid (also known as a server farm). Distributed caches usually can be configured to keep all data objects highly available after a node or a portion of the network fails. This is accomplished by replicating the objects to other nodes and using the replicated copies, if necessary, to recover from a node failure. In addition, distributed caches usually can be configured to automatically distribute data objects among the nodes in the computation grid so that they are evenly spread across all servers. A popular method for doing this is to separate the data objects into groups of objects called partitions and to evenly distribute the partitions among the nodes. The number of partitions can be much larger than the number of nodes. (In general, the use of partitions to distribute load among the nodes of a computational grid has been well established in prior art.) An example of a partitioned, distributed data cache is shown in FIG. 3. If a node fails, the cache's partitions are re-distributed among the surviving nodes, as shown in FIG. 4, and the objects remain assigned to their original partitions; the number of partitions remains unchanged.
Combined with reliable multicast from distributed computing systems, a distributed cache forms an excellent basis for completing steps 1 and a portion of step 2 within a data parallel operation as described on page 2. First, the data sets to be operated on by the method can be stored in the distributed cache as data objects, which the cache has evenly distributed among the partitions of the cache and thereby among the nodes of the grid. Once a method invocation has been reliably multicast to all nodes in step 1, each node can invoke the method on the data objects that the distributed cache has placed on the local node (assuming that the distributed cache provides a means for doing this). Since the data objects are highly available, the data sets are not lost if a node fails or becomes inaccessible due to a partial network failure. Also, after a failure, the distributed cache redistributes all objects as necessary among the surviving nodes.
However, this combination of prior art does not fully implement steps 2 and 3 of a data parallel operation, and so it does not make the overall data parallel operation highly available. After a failure occurs, the prior art does not solve the problems of ensuring that the method has been successfully invoked on all specified data objects and that the completion of the data parallel operation is communicated to the originating node (or its successor after a failure).